<?php

require './vendor/autoload.php';

$config = array(
    "es_hosts" => "192.168.166.35:9200",
    "es_index" => "fkl_ceshi",
    "es_type" => "test_type",
    "mongo_hosts" => 'mongodb://192.168.112.17:27017',
    "mongo_database" => "fkl",
    "mongo_tables" => array("win_cdr"),
    "csv_path" => "1.csv",
    "max_num" => 10,   //每次最多删除条数
    "vcc_id" => 0,
);

//defaultLogge是一个快捷的方法；你也可以自己通过Monolog来创建自己的logger;
//$logger = \Elasticsearch\ClientBuilder::defaultLogger("./logs/test.log");
$logger = new \Monolog\Logger("test");
$logger->pushHandler(new \Monolog\Handler\StreamHandler("./logs/test.log", \Monolog\Logger::WARNING));
/** @var \Elasticsearch\Connections\Connection $esClient */
$esClient = \Elasticsearch\ClientBuilder::create()
    ->setHosts(array($config['es_hosts']))
    ->setLogger($logger)
    ->build();
//连接mongodb
$client = new MongoDB\Client($config['mongo_hosts']);
$mongoClient = $client->selectDatabase($config['mongo_database']);

$nowNum = $config['max_num'];
$param = array();
//打开文件
$csvFile = fopen($config['csv_path'],'r');
while ($csvData = fgetcsv($csvFile)) { //每次读取CSV里面的一行内容
    //过滤表头数据
    if(!is_numeric($csvData[1])) {
        continue;
    }
    $param[] = array(
        "bool" => array(
            "must" => array(
                array(
                    "match" => array("start_time" => (int) strtotime($csvData[5]))
                ),
                array(
                    "match" => array("caller" => (string) $csvData[1])
                )
            )
        )
    );
    if (--$nowNum <= 0) {
        exce($config, $param, $esClient, $mongoClient);
        $nowNum = $config['max_num'];
        $param = array();
    }
}
if (!empty($param)) {
    exce($config, $param, $esClient, $mongoClient);
}

/**
 * @param $config
 * @param $param
 * @param \Elasticsearch\Client $esClient
 * @param \MongoDB\Client $mongoClient
 */
function exce($config, $param, $esClient, $mongoClient)
{
    //先查询 es
    $esCondition = array(
        'index' => $config["es_index"],
        'type' => $config["es_type"],
        'body'=>array(
            'query'=>array(
                "bool" => array(
                    "must" => array(
                        array(
                            'match'=>array(
                                'vcc_id' => $config['vcc_id'],
                            )
                        )
                    ),
                    "should" => $param
                )
            )
        ),
    );
    $response = $esClient->search($esCondition);
    if (empty($response['hits']['hits'])) {
        return;
    }
    $sources = $response['hits']['hits'];
    $ids = array();
    $_ids = array();
    foreach ($sources as $source) {
        $ids[] = (int) $source['_source']['id'];
        $_ids[] = (string) $source['_id'];
    }
    $mgCondition = array(
        "_id" => array('$in' => $ids)
    );
    //循环删除所有mongodb表
    foreach ($config['mongo_tables'] as $val) {
        $col = $mongoClient->selectCollection($val);
        $col->deleteMany($mgCondition);
    }
    //删除es
    foreach ($_ids as $val) {
        $esCondition = array(
            'index' => $config["es_index"],
            'type' => $config["es_type"],
            "id" => $val,
        );
        $esClient->delete($esCondition);
    }
}
